/*
**    Copyright (C) 2003-2011 Institute for Systems Biology
**                            Seattle, Washington, USA.
**
**    This library is free software; you can redistribute it and/or
**    modify it under the terms of the GNU Lesser General Public
**    License as published by the Free Software Foundation; either
**    version 3 of the License, or (at your option) any later version.
**
**    This library is distributed in the hope that it will be useful,
**    but WITHOUT ANY WARRANTY; without even the implied warranty of
**    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
**    Lesser General Public License for more details.
**
**    You should have received a copy of the GNU Lesser General Public
**    License along with this library; if not, write to the Free Software
**    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307  USA
*/

package org.systemsbiology.rface.hadoop.data;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.*;

import org.systemsbiology.util.KahanSummation;
import org.systemsbiology.util.SortablePair;
import org.systemsbiology.util.Summator;

import org.apache.hadoop.io.Writable;

public class TargetRecord implements Writable{

	private Map<String, ArrayList<SortablePair<Double, String>>> data;
	
	private static final int MAGIC_OPEN = 0x1BADFACE;
	private static final int MAGIC_CLOSE = 0xFEDDACA7;
	
	public TargetRecord(){
		data = new HashMap<String, ArrayList<SortablePair<Double, String>>>();
	}
	
	public TargetRecord(TargetRecord duplicate){
		data = new HashMap<String, ArrayList<SortablePair<Double, String>>>();
		for(Map.Entry<String, ArrayList<SortablePair<Double, String>>> record : duplicate.data.entrySet()){
			ArrayList<SortablePair<Double, String>> pushMe = new ArrayList<SortablePair<Double, String>>();
			for(SortablePair<Double, String> value : record.getValue()){
				pushMe.add(new SortablePair<Double, String>(value));
			}
			data.put(record.getKey(), pushMe);
		}
	}
	
	public void addPredictor(String filename, String identifier, double value){
		if(!data.containsKey(filename)){
			data.put(filename, new ArrayList<SortablePair<Double, String>>());
		}
		data.get(filename).add(new SortablePair<Double, String>(value, identifier));
	}
	
	public void sortAll(){
		for(ArrayList<SortablePair<Double, String>> list : data.values()){
			Collections.sort(list);
		}
	}
	
	public void mergeIn(TargetRecord rec){
		for(Map.Entry<String, ArrayList<SortablePair<Double, String>>> fileGroup : rec.data.entrySet()){
			if(!data.containsKey(fileGroup.getKey())){
				data.put(fileGroup.getKey(), new ArrayList<SortablePair<Double, String>>());
			}
			ArrayList<SortablePair<Double, String>> target = data.get(fileGroup.getKey());
			for(SortablePair<Double, String> line : fileGroup.getValue()){
				target.add(new SortablePair<Double, String>(line));
			}
		}
	}
	
	/**
	 * Result type for inconsistentcyScore. Basically a struct.
	 */
	public static class InconsistencyResult implements Comparable<InconsistencyResult>{
		public double score;
		public HashMap<String, Double> scaledStddevs;
		public HashMap<String, Double> avgRanks;
		public HashMap<String, int[]> ranks;
		
		public InconsistencyResult(
				double score,
				HashMap<String, Double> scaledStddevs,
				HashMap<String, Double> avgRanks,
				HashMap<String, int[]> ranks){
			this.score = score;
			this.scaledStddevs = scaledStddevs;
			this.avgRanks = avgRanks;
			this.ranks = ranks;
		}
		
		public int compareTo(InconsistencyResult o){
			int dcomp = Double.compare(score, o.score);
			if(dcomp != 0) return dcomp;
			ArrayList<String> str1 = new ArrayList<String>();
			str1.addAll(scaledStddevs.keySet());
			ArrayList<String> str2 = new ArrayList<String>();
			str2.addAll(scaledStddevs.keySet());
			if(str1.size() < str2.size()){
				return -1;
			}
			if(str1.size() > str2.size()){
				return 1;
			}
			Collections.sort(str1);
			Collections.sort(str2);
			for(int k = 0; k < str1.size(); ++k){
				int retCandidate = str1.get(k).compareTo(str2.get(k));
				if(retCandidate != 0) return retCandidate;
			}
			return 0;
		}
		
		public boolean equals(Object o){
			if(o == null) return false;
			if(! getClass().equals(o.getClass())) return false;
			return compareTo((InconsistencyResult)o) == 0;
		}
		
		public int hashCode(){
			ArrayList<String> strs = new ArrayList<String>();
			strs.addAll(scaledStddevs.keySet());
			Collections.sort(strs);
			return Double.valueOf(score).hashCode() ^ strs.hashCode();
		}
		
		private static String pyDict(Map<String, Double> dict){
			StringBuilder dump = new StringBuilder("{");
			for(Map.Entry<String, Double> entry : dict.entrySet()){
				dump.append("\"");
				dump.append(entry.getKey());
				dump.append("\":");
				dump.append(entry.getValue());
				dump.append(","); //trailing comma is legal in Python
			}
			dump.append("}");
			return dump.toString();
		}
		
		private static String pyListDict(Map<String, int[]> dict){
			StringBuilder dump = new StringBuilder("{");
			for(Map.Entry<String, int[]> entry : dict.entrySet()){
				dump.append("\"");
				dump.append(entry.getKey());
				dump.append("\":");
				dump.append(Arrays.toString(entry.getValue()));
				dump.append(","); //trailing comma is legal in Python
			}
			dump.append("}");
			return dump.toString();
		}
		
		public String toString(){
			//score, deviations, average ranks, ranks; use Python format
			return String.valueOf(score) + "\t" + pyDict(scaledStddevs) +
					"\t" + pyDict(avgRanks) + "\t" + pyListDict(ranks);
		}
	}
	
	public InconsistencyResult inconsistencyScore(int relevantHead, int pessimism){
		sortAll();
		
		HashSet<String> featuresOfNote = new HashSet<String>();
		int maxK = -1;
		int maxLen = -1;
		for(ArrayList<SortablePair<Double, String>> list : data.values()){
			if(list.size() > maxLen) maxLen = list.size();
			for(int k = 0; k < relevantHead && k < list.size(); ++k){
				if(k > maxK) maxK = k; //in case all lists are short, we want to find our furthest index
				featuresOfNote.add(list.get(k).second);
			}
		}
		if(featuresOfNote.isEmpty()){ //it may not be very useful, but at least it's consistent
			return new InconsistencyResult(0.0, 
					new HashMap<String, Double>(),
					new HashMap<String, Double>(),
					new HashMap<String, int[]>());
		}
		
		int fail = maxLen + pessimism;
		
		HashMap<String, int[]> ranks = new HashMap<String, int[]>(featuresOfNote.size() * 2);
		for(String s: featuresOfNote){
			int[] valRanks = new int[data.size()];
			Arrays.fill(valRanks, fail);
			ranks.put(s, valRanks);
		}
		
		int i = 0;
		for(ArrayList<SortablePair<Double, String>> records: data.values()){
			int rank = 0;
			for(SortablePair<Double, String> pair : records){
				if(ranks.containsKey(pair.second)){
					ranks.get(pair.second)[i] = rank;
				}
				++rank;
			}
			++i;
		}
		
		HashMap<String, Double> averageRanks = new HashMap<String, Double>(ranks.size() * 2);
		for(Map.Entry<String, int[]> rollup : ranks.entrySet()){
			int total = 0;
			for(int k: rollup.getValue()){
				total += k;
			}
			averageRanks.put(rollup.getKey(), (double)total/(double)rollup.getValue().length);
		}
		
		HashMap<String, Double> stddevs = new HashMap<String, Double>(ranks.size() * 2);
		for(Map.Entry<String, int[]> rollup: ranks.entrySet()){
			double avg = averageRanks.get(rollup.getKey());
			Summator variance = new KahanSummation();
			for(int n: rollup.getValue()){
				double diff = avg - (double)n;
				diff /= (maxK + 1);
				variance.add(diff * diff);
			}
			stddevs.put(rollup.getKey(), Math.sqrt(variance.total()));
		}
		
		Summator totalStddev = new KahanSummation();
		for(double s : stddevs.values()){
			totalStddev.add(s);
		}
		double dispersion = totalStddev.total()/(maxK + 1); //dispersion: it is not an average
		
		return new InconsistencyResult(
				dispersion,
				stddevs,
				averageRanks,
				ranks);
	}
	
	public void readFields(DataInput arg0) throws IOException {
		data = new HashMap<String, ArrayList<SortablePair<Double, String>>>();
		if(arg0.readInt() != MAGIC_OPEN){
			throw new IOException("Opening signature check failure!");
		}
		int nFiles = arg0.readInt();
		for(int i = 0; i < nFiles; ++i){
			String filename = arg0.readUTF();
			int nEntries = arg0.readInt();
			ArrayList<SortablePair<Double, String>> list = new ArrayList<SortablePair<Double, String>>(nEntries);
			for(int j = 0; j < nEntries; ++j){
				Double value = arg0.readDouble();
				String predictor = arg0.readUTF();
				list.add(new SortablePair<Double, String>(value, predictor));
			}
			data.put(filename, list);
		}
		if(arg0.readInt() != MAGIC_CLOSE){
			throw new IOException("Closing signature check failure! Compare serialization/deserialization algorithms");
		}
	}

	public void write(DataOutput arg0) throws IOException {
		arg0.writeInt(MAGIC_OPEN);
		arg0.writeInt(data.size());
		for(Map.Entry<String, ArrayList<SortablePair<Double, String>>> entry: data.entrySet()){
			arg0.writeUTF(entry.getKey());
			ArrayList<SortablePair<Double, String>> list = entry.getValue();
			arg0.writeInt(list.size());
			for(SortablePair<Double, String> thing : list){
				arg0.writeDouble(thing.first);
				arg0.writeUTF(thing.second);
			}
		}
		arg0.writeInt(MAGIC_CLOSE);
		
	}

}
